package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo23CheckPoint {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo23CheckPoint")

    val sc: SparkContext = new SparkContext(conf)
    sc.setCheckpointDir("spark/data/ck")

    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")

    val stuMapRDD: RDD[String] = stuRDD.map(line => {
      println("进入了map算子")
      line
    })

    // 在使用checkpoint之前需要先通过sc设置目录
    /**
     * checkpoint同缓存类似，但缓存是出于性能考虑的，ck是出于可靠性考虑的
     * ck一般会将数据保存到可靠的文件系统中，例如HDFS
     * cache是尽可能将数据保存到内存中
     *
     * ck在保存的时候需要单独触发一个Job完成，会从头开始把需要做ck的RDD计算出来并进行保存
     * 相当于做ck的RDD被使用了多次，所以可以进行cache缓存来提高效率，这样ck就不要再从头计算了
     *
     * checkpoint实际上一般用于实时程序
     */
    stuMapRDD.cache()
    stuMapRDD.checkpoint()

    // 统计班级人数
    val clazzCntRDD: RDD[(String, Int)] = stuMapRDD.map(line => (line.split(",")(4), 1)).reduceByKey(_ + _)
    clazzCntRDD.foreach(println)

    // 统计性别人数
    val genderCntRDD: RDD[(String, Int)] = stuMapRDD.map(line => (line.split(",")(3), 1)).reduceByKey(_ + _)
    genderCntRDD.foreach(println)

    // 统计年龄分布情况
    val ageCntRDD: RDD[(String, Int)] = stuMapRDD.map(line => (line.split(",")(2), 1)).reduceByKey(_ + _)
    ageCntRDD.foreach(println)

    while(true){

    }


  }

}
